package com.atguigu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

//状态
//有状态操作.
object SprakStreaming07_State {
  def main(args: Array[String]): Unit = {

    //所谓的状态，其实就是将数据进行保存
    //默认情况下。SparkStreaming的操作是无状态的.
    //统计结果是周期的，只对当前周期有效.

    val conf: SparkConf = new SparkConf().setAppName("tarnsfrom").setMaster("local[*]")
    val ssc = new StreamingContext(conf,Seconds(3))


    //从网络端口获取数据,拿到一行字符串
    val socketData: ReceiverInputDStream[String] = ssc.socketTextStream("localhost",9999)
    //将数据进行扁平化处理
    val words: DStream[String] = socketData.flatMap(_.split(" "))
    //将数据结构进行转换
    val wordToOne: DStream[(String, Int)] = words.map((_,1))
    //聚合操作(value1+value2)
    val wordToCount: DStream[(String, Int)] = wordToOne.reduceByKey(_ + _)
    wordToCount.print()

    ssc.start()
    ssc.awaitTermination()


  }

}
